package com.cloud.message;

import com.cloud.exception.CustomException;
import com.cloud.utils.JSONUtils;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.HashMap;
import java.util.Map;

/**
 * Description：RocketMQ 消费抽象基类
 *
 * @author yzj
 */
public abstract class AbstractMQConsumer<T> {

    private static Logger logger = LoggerFactory.getLogger(AbstractMQConsumer.class);

    /**
     * 反序列化解析消息
     *
     * @param message 消息体
     * @return 序列化结果
     */
    protected T parseMessage(MessageExt message) {
        if (ObjectUtils.isEmpty(message) || ObjectUtils.isEmpty(message.getBody())) {
            return null;
        }
        final Type type = this.getMessageType();
        if (type instanceof Class) {
            try {
                return JSONUtils.toObject(new String(message.getBody()), (Class<T>) type);
            } catch (Exception e) {
                logger.error("解析消息对象失败:{}", e.getMessage());
            }
        } else {
            logger.warn("解析消息错误.{}", message);
        }
        return null;
    }

    protected Map<String, Object> parseExtParam(MessageExt message) {
        Map<String, Object> extMap = new HashMap<>();
        // parse message property
        extMap.put(MessageExtConst.PROPERTY_TOPIC, message.getTopic());
        extMap.putAll(message.getProperties());
        // parse messageExt property
        extMap.put(MessageExtConst.PROPERTY_EXT_BORN_HOST, message.getBornHost());
        extMap.put(MessageExtConst.PROPERTY_EXT_BORN_TIMESTAMP, message.getBornTimestamp());
        extMap.put(MessageExtConst.PROPERTY_EXT_COMMIT_LOG_OFFSET, message.getCommitLogOffset());
        extMap.put(MessageExtConst.PROPERTY_EXT_MSG_ID, message.getMsgId());
        extMap.put(MessageExtConst.PROPERTY_EXT_PREPARED_TRANSACTION_OFFSET, message.getPreparedTransactionOffset());
        extMap.put(MessageExtConst.PROPERTY_EXT_QUEUE_ID, message.getQueueId());
        extMap.put(MessageExtConst.PROPERTY_EXT_QUEUE_OFFSET, message.getQueueOffset());
        extMap.put(MessageExtConst.PROPERTY_EXT_RECONSUME_TIMES, message.getReconsumeTimes());
        extMap.put(MessageExtConst.PROPERTY_EXT_STORE_HOST, message.getStoreHost());
        extMap.put(MessageExtConst.PROPERTY_EXT_STORE_SIZE, message.getStoreSize());
        extMap.put(MessageExtConst.PROPERTY_EXT_STORE_TIMESTAMP, message.getStoreTimestamp());
        extMap.put(MessageExtConst.PROPERTY_EXT_SYS_FLAG, message.getSysFlag());
        extMap.put(MessageExtConst.PROPERTY_EXT_BODY_CRC, message.getBodyCRC());
        return extMap;
    }

    /**
     * 解析消息类型
     *
     * @return 消息类型
     */
    protected Type getMessageType() {
        Type superType = this.getClass().getGenericSuperclass();
        if (superType instanceof ParameterizedType) {
            ParameterizedType parameType = (ParameterizedType) superType;
            Type[] actualTypes = parameType.getActualTypeArguments();
            if (actualTypes.length == 1) {
                return actualTypes[0];
            } else {
                throw new CustomException("类型参数的数量必须为1");
            }
        } else {
            // 如果没有定义泛型，解析为Object
            return Object.class;
        }
    }

}
